package com.atguigu.PracticalProject.app.dwd.db;

import com.atguigu.PracticalProject.app.BaseSqlApp;
import com.atguigu.PracticalProject.common.Constant;
import com.atguigu.PracticalProject.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdUserRegister extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdUserRegister().init(2003, 2, "DwdUserRegister");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        // 1. 读取ods_db
        readOdsDb(tEnv, "DwdUserRegister");

        // 2. 读用户表
        Table userInfo = tEnv.sqlQuery("select " +
                "data['id'] user_id, " +
                "data['create_time'] create_time, " +
                "ts  " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='user_info' "
        );
        tEnv.createTemporaryView("user_info", userInfo);

        // 3. kafka创建用户表
        tEnv.executeSql("create table dwd_user_register( " +
            "`user_id` string, " +
            "`date_id` string, " +
            "`create_time` string, " +
            "`ts` bigint " +
            ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_USER_REGISTER));

        //4. 写入kafka
        tEnv.executeSql("insert into dwd_user_register\n" +
            "select \n" +
            "user_id,\n" +
            "date_format(create_time, 'yyyy-MM-dd') date_id,\n" +
            "create_time,\n" +
            "ts\n" +
            "from user_info");
    }
}
